package com.hehao.java_basic.thread;

import com.alibaba.fastjson2.JSON;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.junit.Test;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

/**
 * @author Damon
 * @date 2021/12/20
 * parallelStream使用demo
 */
@Slf4j
public class ParallelStream {
    public static void main(String[] args) {
        List<Person> persons = constructPersons();
        //1.同步for运行
        long start1 = System.currentTimeMillis();
        for (Person p : persons) {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
            }
            System.out.println(p.name);
        }
        long end1 = System.currentTimeMillis();
        System.out.println("同步for耗时:" + (end1 - start1));

        //2.顺序流运行
        long start2 = System.currentTimeMillis();
        persons.stream().forEach(x -> {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
            }
            System.out.println(x.name);
        });
        long end2 = System.currentTimeMillis();
        System.out.println("顺序流耗时:" + (end2 - start2));

        //3.并行流运行
        long start3 = System.currentTimeMillis();
        persons.parallelStream().forEach(o -> {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
            }
            System.out.println(o.name);
        });
        long end3 = System.currentTimeMillis();
        System.out.println("并行流耗时:" + (end3 - start3));
    }

    /**
     * parallelStream线程安全测试
     */
    @Test
    public void parallelStreamSafe() {
        List<Integer> listFor = new ArrayList<>();
        List<Integer> listParallel = new ArrayList<>();

        IntStream.range(0, 1000).forEach(listFor::add);
        IntStream.range(0, 1000).parallel().forEach(listParallel::add);

        System.out.println("listFor size :" + listFor.size());
        System.out.println("listParallel size :" + listParallel.size());
    }

    @Data
    @AllArgsConstructor
    static class Person {
        int id;
        String name;
        String sex;
        float height;
    }

    public static List<Person> constructPersons() {

        List<Person> persons = new ArrayList<Person>();
        for (int i = 0; i < 5; i++) {
            Person p = new Person(i, "name" + i, "sex" + i, i);
            persons.add(p);
        }
        return persons;
    }

    @Test
    public void testParallelStream() {
        List<Person> persons = constructPersons();
        List<CompletableFuture<String>> futures = persons.parallelStream()
                .map(person -> CompletableFuture.supplyAsync(() -> {
                            try {
                                return dealParam(person);
                            } catch (Exception e) {
                                log.error("", e);
                                return null;
                            }
                        })
                )
                .toList();
        CompletableFuture<Void> allFutures = CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]));
        List<String> list = null;
        try {
            // 等待所有任务完成
            //allFutures.join();
            // 设置最长等待时间为2秒
            allFutures.get(2, TimeUnit.SECONDS);
        } catch (InterruptedException | ExecutionException | TimeoutException e) {
            // 处理超时或异常情况
            e.printStackTrace();
        }
        list = futures.stream().filter(CompletableFuture::isDone)
                .map(CompletableFuture::join).toList();
        log.info("=========>json:{}", JSON.toJSONString(list));
    }

    private String dealParam(Person person) throws Exception {
        log.info("====>线程名称{},人名:{}", Thread.currentThread().getName(), person.getName());
        if ("name3".equals(person.getName())) {
//            throw new Exception("错误！");
            Thread.sleep(10000);
        }
        return person.getName();
    }

    private String handleException(Throwable ex, Person person) {
        log.error("+++++>人名:{}", person.getName(), ex);
        return null;
    }
}
